package com.mindata.blockchain.block.queue;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.google.common.base.Preconditions;
import com.mindata.blockchain.block.Instruction;
import com.mindata.blockchain.common.ResultPack;
import com.mindata.blockchain.common.exception.BlockException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 指令池队列操作
 * User: jasperxgwang
 * Date: 2018-7-24 11:59
 */
public class InstructionSender {

    private static final Logger logger = LoggerFactory.getLogger(InstructionSender.class);

    /**
     * 本地队列
     */
    private DataQueue<Instruction> instructionQueue;
    /**
     * 队列中指令hash
     */
    private Set<String> instructionHash;

    /**
     * 最大指令池，-1 默认不限制
     */
    private int capital = -1;

    /**
     * 单个区块最大打包指令数 -1 默认不限制
     */
    private int packageLimitSize = -1;
    /**
     * 队列消费回调，触发生成block
     */
    private SenderCallBack senderCallBack;

    /**
     * 初始化
     *
     * @param group            队列标识
     * @param capital          队列容量
     * @param packageLimitSize 区块打包指令限制
     * @param senderCallBack   队列消费回调
     */
    public InstructionSender(final String group, int capital, int packageLimitSize, SenderCallBack senderCallBack) {
        logger.info("init InstructionSender group:[{}] capital:[{}] packageLimitSize:[{}]", group, capital, packageLimitSize);
        Preconditions.checkNotNull(senderCallBack, "senderCallBack is null");
        this.senderCallBack = senderCallBack;
        this.capital = capital;
        this.packageLimitSize = packageLimitSize;

        //本地队列
        instructionQueue = new DataQueue(capital);
        instructionHash = new ConcurrentHashSet();

        //消费线程
        ExecutorService poll = Executors.newFixedThreadPool(1, r -> {
            Thread t = new Thread(r, "group-" + group);
            return t;
        });
        poll.execute(new Consume());

        //JVM停止或者重启时，清空队列
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                execute();
            } catch (Exception e) {
                logger.error("jvm shutdown hook error", e);
            }
        }));
    }

    private InstructionSender() {
    }

    /**
     * 加入指令池
     *
     * @param instruction
     * @return
     */
    public ResultPack<String> production(Instruction instruction) {
        //判断指令是否已存在
        String key = instruction.getHash();
        if (instructionHash.contains(key)) {
            return ResultPack.failed("[" + key + "]已经存在指令池中");
        }

        ResultPack<String> resultPack = instructionQueue.offer(instruction);
        //入队失败
        if (resultPack.isFailed()) {
            throw new BlockException("指令池已满，当前队列数量:" + instructionQueue.getSize() + " 队列最大限制:" + capital);
        } else {
            instructionHash.add(key);
        }
        return resultPack;
    }


    private void execute() {
        if (packageLimitSize <= 0) {
            packageLimitSize = instructionQueue.getSize();
        }
        List<Instruction> pollData = instructionQueue.pollList(packageLimitSize);
        if (!CollectionUtils.isEmpty(pollData)) {
            for (Instruction ins : pollData) {
                instructionHash.remove(ins.getHash());
            }
            try {
                senderCallBack.consume(pollData);
            } catch (Exception e) {
                logger.error("exec consume error", e);
            }
        }
    }


    /**
     * 消费线程
     */
    class Consume implements Runnable {

        int notFullQueueFlag = 0;

        @Override
        public void run() {
            logger.info("InstructionSender start");
            while (true) {
                try {
                    if (instructionQueue.isEmpty()) {
                        sleep(2000);
                        logger.debug("instructionQueue no data sleep 2s.");
                    } else {
                        long queueSize = instructionQueue.getSize();
                        if (queueSize >= capital * 0.8) {
                            //满足队列的80%,马上消费,否则休眠50ms,待队列聚合
                            execute();
                        } else {
                            if (notFullQueueFlag % 10 == 0) {
                                //不满足80%,则10次，即 10*300=3000 millis 清空一次队列
                                execute();
                                notFullQueueFlag = 0;
                                logger.warn("the queue(capital:{} queueSize:{}) is not full ,sleep 3000 millis", capital, queueSize);
                            }
                            notFullQueueFlag++;
                            sleep(300);
                        }
                    }
                } catch (Exception e) {
                    logger.error("InstructionSender error:", e);
                }
            }
        }
    }

    public static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            logger.error("sleep ex:", e);
        }
    }
}
